package com.dongdongshop.mq;

import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.mapper.TbUserScoreAccountMapper;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.pojo.TbUserScoreAccountExample;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Component
@RocketMQMessageListener(consumerGroup = "UserScoreTable_Consumer",topic = "AddUserScoreTable")
public class UserScoreAccountMQ implements RocketMQListener<Map>, RocketMQPushConsumerLifecycleListener {

    @Resource
    private TbUserScoreAccountMapper tbUserScoreAccountMapper;

    @Override
    public void onMessage(Map map) {
        //使用ack,此方法自动失效
        System.out.println("onMessage");
    }

    //mq的ACK机制
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        System.out.println("ACK机制");
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                try {
                    MessageExt msg = list.get(0);
                    String s = new String(msg.getBody(), "UTF-8");
                    System.out.println(s);
//                    String msgbody = new String(msg.getBody(), "utf-8");
//                    System.out.println(msgbody + " Receive New Messages: " + list);
                    Map<String,Object> map = JSONObject.parseObject(s,Map.class);
                    /*执行tb_user_score_account表的第一次数据添加*/
                    String userName = (String)map.get("userName");
                    Double money = Double.valueOf((String) map.get("money"));
                    TbUserScoreAccountExample example = new TbUserScoreAccountExample();
                    TbUserScoreAccountExample.Criteria criteria = example.createCriteria();
                    criteria.andUserIdEqualTo(userName);
                    List<TbUserScoreAccount> tbUserScoreAccounts = tbUserScoreAccountMapper.selectByExample(example);
                    if (tbUserScoreAccounts.size() == 0){
                        TbUserScoreAccount scoreAccount = new TbUserScoreAccount();
                        scoreAccount.setUserId(userName);
                        scoreAccount.setTotalIntegral(money);
                        scoreAccount.setConsumptionIntegral(money);
                        scoreAccount.setCreationTime(new Date());
                        scoreAccount.setUpdateTime(new Date());
                        tbUserScoreAccountMapper.insert(scoreAccount);
                    }else {
                        TbUserScoreAccount score = tbUserScoreAccounts.get(0);
                        score.setUpdateTime(new Date());
                        score.setTotalIntegral(score.getTotalIntegral() + money);
                        tbUserScoreAccountMapper.updateByPrimaryKeySelective(score);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    // 该条消息可以存储到DB或者LOG日志中，或其他处理方式，然后重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
